package com.xzx.mr.mapjoin;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

/**
 * 将小表缓存在内存中
 *
 * @author xinzhixuan
 * @version V1.0
 * @date 2019/7/26 22:32
 */
public class DistributedCacheMapper extends Mapper<LongWritable, Text, Text, NullWritable> {

    private Text outputKey = new Text();

    private Map<String, String> product = new HashMap<>();

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        URI[] cacheFiles = context.getCacheFiles();
        String path = cacheFiles[0].getPath();
        BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(path),
                StandardCharsets.UTF_8));
        String line;
        while (StringUtils.isNotBlank(line = reader.readLine())) {
            //01	小米
            String[] split = line.split("\t");
            product.put(split[0], split[1]);
        }
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 1001	01	1
        String line = value.toString();
        String[] split = line.split("\t");

        outputKey.set(split[0] + "\t" + product.get(split[1]) + "\t" +split[2]);

        context.write(outputKey, NullWritable.get());
    }
}
